/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package br.ufba.lasid.comm;

import br.ufba.lasid.comm.api.IReceiver;
import br.ufba.lasid.comm.api.ICommunicator;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 *
 * @author aliriosa
 */
public class Communicator implements ICommunicator {

    protected static final String AllComponents = "all";
    private static final String ExchangeType = "topic";
    private static final String at = "@";
    private Receivertable receivertable = new Receivertable();
    private static final Log logger = LogFactory.getLog(Communicator.class);

    @Override
    public void register(IReceiver receiver, String hostID, String componentID) throws Exception {

        String bcastkey = getRoutingKey(hostID);
        String desiredKey = getRoutingKey(hostID, componentID);

        IReceiver registerReceiver = receivertable.get(bcastkey);

        if (registerReceiver != null) {
//            logger.warn(
//              "Receiver " + receiver + " is already registered in " + bcastkey + ", " + 
//              "so it couldn't be registered in " + desiredKey
//             );
            return;
        }

        receivertable.put(desiredKey, receiver);

//        logger.info(
//          "Receiver " + receiver + " was successfully registered in " + desiredKey
//         );

        startReceptionService(hostID);

    }

    public void startReceptionService(String hostID) throws Exception {
        new Receptor(hostID).start();
    }

    @Override
    public void register(IReceiver receiver, String hostID) throws Exception {
        register(receiver, hostID, AllComponents);
    }

    @Override
    public void deregister(String hostID, String componentID) throws Exception {
        String desiredEntry = getRoutingKey(hostID, componentID);
        receivertable.remove(desiredEntry);
    }

    @Override
    public void deregister(String hostID) throws Exception {
        deregister(hostID, AllComponents);
    }

    @Override
    public void send(String hostID, String componentID, String message) throws Exception {

        String routingkey = getRoutingKey(hostID, componentID);

        ConnectionFactory factory = new ConnectionFactory();

        factory.setHost(hostID);

        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();

        channel.exchangeDeclare(hostID, ExchangeType);

        channel.basicPublish(hostID, routingkey, null, message.getBytes());

        //logger.info(message + " has been sent to " + componentID + "@" + hostID);        

        channel.close();
        connection.close();

    }

    @Override
    public void send(String hostID, String message) throws Exception {
        send(hostID, AllComponents, message);
    }

    public String getRoutingKey(String hostID, String componentID) {
        return componentID + at + hostID;
    }

    public String getHostIDFromRoutingKey(String routingKey) {
        int i = routingKey.indexOf(at) + 1;
        return routingKey.substring(i);
    }

    public String getComponentIDFromRoutingKey(String routingKey) {
        int i = routingKey.indexOf(at);
        return routingKey.substring(0, i);
    }

    public String getRoutingKey(String hostID) {
        return AllComponents + at + hostID;
    }

    @Override
    public void run() {
        //do nothing
    }

    @Override
    public void startup() throws Exception {
        //do nothing
    }

    @Override
    public String getName() throws Exception {
        return "Communicator";
    }

    class Receptor extends Thread {

        private String hostID;

        public Receptor(String hostID) {
            this.hostID = hostID;
        }

        @Override
        public void run() {

            try {
                ConnectionFactory factory = new ConnectionFactory();

                factory.setHost(hostID);

                Connection connection = factory.newConnection();

                Channel channel = connection.createChannel();

                channel.exchangeDeclare(hostID, ExchangeType);

                String queueName = channel.queueDeclare().getQueue();

                channel.queueBind(queueName, hostID, "#");

                QueueingConsumer consumer = new QueueingConsumer(channel);

                channel.basicConsume(queueName, true, consumer);

                logger.info(
                        "A new MDAReceivingService has been initiated "
                        + "to receive messages from" + queueName);

                while (true) {

                    try {
                        QueueingConsumer.Delivery delivery = consumer.nextDelivery();

                        String message = new String(delivery.getBody());
                        String routing = delivery.getEnvelope().getRoutingKey();
                        String component = getComponentIDFromRoutingKey(routing);

                        IReceiver receiver = receivertable.get(routing);

                        if (receiver == null) {
                            receiver = receivertable.get(getRoutingKey(hostID));
                        }

                        if (receiver != null) {
                            receiver.receive(hostID, component, message);
                        }

                    } catch (Exception e) {
                        //logger.error("Exception detected during reception", e);
                    }

                }
            } catch (Exception e) {
                logger.error("Exception detected during receiver initialization", e);
            }
        }
    }
}
